package demo.booleanteeth.rxjavademo;


import android.util.Log;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.io.Serializable;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;

public class Test {

    private static final String TAG = "Test";

    public static void main(String[] args) {
        Flowable<Module> flowable = Flowable.create(new FlowableOnSubscribe<Module>() {
            @Override
            public void subscribe(FlowableEmitter<Module> emitter) throws Exception {
                System.out.println("emit 1");
                emitter.onNext(new Module());
                System.out.println("emit 2");
                emitter.onNext(new Module());
                System.out.println("emit 3");
                emitter.onNext(new Module());
                System.out.println("emit complete");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR);
        Flowable<Module> flowable2 = Flowable.create(new FlowableOnSubscribe<Module>() {
            @Override
            public void subscribe(FlowableEmitter<Module> e) throws Exception {

            }
        }, BackpressureStrategy.ERROR);


        Flowable.concatArray(flowable, flowable).subscribe(new Consumer<Module>() {
            @Override
            public void accept(Module module) throws Exception {

            }
        });
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                 Thread.sleep(1000);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Thread.sleep(1000);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Thread.sleep(1000);
                Log.d(TAG, "emit 4");
                emitter.onNext(4);
                Thread.sleep(1000);
                Log.d(TAG, "emit complete1");
                emitter.onComplete();
            }
        }); Observable<String> observable2 = Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "emit A");
                        emitter.onNext("A");
                        Thread.sleep(1000);
                        Log.d(TAG, "emit B");
                        emitter.onNext("B");
                        Thread.sleep(1000);
                        Log.d(TAG, "emit C");
                        emitter.onNext("C");
                        Thread.sleep(1000);
                        Log.d(TAG, "emit complete2");
                        emitter.onComplete();
                    }
                });
                Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
                    @Override
                    public String apply(Integer integer, String s) throws Exception {
                        return integer + s;
                    }
                }).subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(TAG, "onSubscribe");
                    }

                    @Override
                    public void onNext(String value) {
                        Log.d(TAG, "onNext: " + value);
                    }


                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "onError");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });
    }

}
